// Copyright (c) 2023 Presto Labs Pte. Ltd.
// Author: fengyang

#include "coin2/exchange/mercado/feed_v4/processor.h"

#include <glog/logging.h>
#include <fmt/format.h>

#include "coin2/base/string_util.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/base/symbology/util.h"
#include "coin2/exchange/mercado/symbology/product.h"
#include "coin2/exchange/mercado/feed_v4/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::mercado::feed_v4::impl {

void MercadoFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  if (packet.payload.empty()) return;

  simdjson::padded_string buffer(packet.payload);

  upd_mgr_.onBeginPacket(packet);

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(packet.payload);
  }

  simdjson::ondemand::document obj;

  if (auto err = parser_.iterate(buffer).get(obj); err) {
    if (parser_config_.continue_on_error()) {
      LOG(ERROR) << "[MercadoProcessor] unknown raw msg: " << packet.payload;
    } else {
      throw std::runtime_error(std::string("JSON parse error: ") + simdjson::error_message(err));
    }
    return;
  }

  static constexpr std::string_view kline_url("candles?");

  std::string_view ch;
  std::string_view url;
  if (auto err = obj["url"].get_string().get(url); !err) {
    if (url.rfind(kline_url) != std::string::npos) {
      ch = "klines_rest";
      const std::string_view interval = "&resolution=";
      size_t pos = url.find(interval);
      size_t sep = url.find("&", pos + 1);
      std::string interval_channel =  [&]()->std::string{auto ret = url.substr(pos, sep - pos -1); return {ret.begin(), ret.end()};}();
      if (!CheckChannelFilter("?symbol={}" + interval_channel)) {
        return;
      }
    }
  }

  std::string_view type_name;
  if (auto err = obj["type"].get_string().get(type_name); !err) {
    ch = type_name;
    if (!CheckChannelFilter(std::string(ch))) {
      return;
    }
    std::string_view native_symbol = obj["id"].get_string();
    const std::string native_symbol_str (native_symbol.begin(), native_symbol.end());
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      return;
    }

    const int64_t ts = obj["ts"].get_int64();
    simdjson::ondemand::object msg = obj["data"].get_object();
    if (ch == "trade") {
      auto* trade = upd_mgr_.GetTradeMutable();
      MercadoFeedUpdateParser::ParseTrade(msg, trade, ts);
      upd_mgr_.PublishTrade(native_symbol_str, false);
    } else if (ch == "orderbook") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        MercadoFeedUpdateParser::ParseBook(
            book,
            msg,
            ts);
        upd_mgr_.PublishBookWithExchangeTimeCheck();
      }
    } else {
      LOG(INFO) << "[MercadoProcessor] meta msg: " << packet.payload;
    }
  } else if (ch == "klines_rest") {
    auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);

    std::string native_symbol = symbology::MercadoProduct::RestToNativeProduct(parameters["symbol"]);
    std::string kline_interval = parameters["resolution"];
    auto interval = MercadoFeedUpdateParser::ParseKlineInterval(kline_interval);
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines != nullptr) {
      simdjson::ondemand::object data = obj["data"].get_object();
      std::optional<int64_t> first_open_timestamp;
      first_open_timestamp = data.at_pointer("/t/0").get<int64_t>() * 1'000'000'000LL;
      data.reset();

      if (klines->Size() == 0) {
        klines->SnapshotClear();
      } else if (first_open_timestamp &&
                 first_open_timestamp < klines->RBegin()->open_timestamp) {
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      bool to_publish = MercadoFeedUpdateParser::ParseKlineRest(
          data, klines, interval, packet.timestamp);
      if (to_publish) upd_mgr_.PublishKline(native_symbol);
    }
  } else {
    LOG(INFO) << "[MercadoProcessor] meta raw msg: " << packet.payload;
  }

  upd_mgr_.onEndPacket(packet);
}  // class MercadoFeedProcessor

}  // namespace coin2::exchange::mercado::feed_v4::impl
